package org.xq.softcup.mq;

import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.xq.softcup.mq.model.XQRegistryDataParamVO;
import org.xq.softcup.mq.util.BasicJson;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
@Slf4j
public class XQRegistryClient {
    private volatile Set<XQRegistryDataParamVO> registryData = new HashSet();
    private volatile ConcurrentMap<String, TreeSet<String>> discoveryData = new ConcurrentHashMap();
    private Thread registryThread;
    private Thread discoveryThread;
    private volatile boolean registryThreadStop = false;
    private XQRegistryBaseClient registryBaseClient;

    public XQRegistryClient(String adminAddress, String accessToken, String biz, String env) {
        this.registryBaseClient = new XQRegistryBaseClient(adminAddress, accessToken, biz, env);
        log.info("XQRegistryClient初始化 [adminAddress={}, accessToken={}, biz={}, env={}]", new Object[]{adminAddress, accessToken, biz, env});

        this.registryThread = new Thread(new Runnable() {
            public void run() {
            while (!XQRegistryClient.this.registryThreadStop) {
                try {
                    if (XQRegistryClient.this.registryData.size() > 0) {
                        boolean ret = XQRegistryClient.this.registryBaseClient.registry(new ArrayList(XQRegistryClient.this.registryData));
                        XQRegistryClient.log.debug("刷新注册数据 {}, registryData = {}", ret ? "成功" : "失败", XQRegistryClient.this.registryData);
                    }
                } catch (Exception e) {
                    if (!XQRegistryClient.this.registryThreadStop) {
                        XQRegistryClient.log.error("注册线程异常", e);
                    }
                }
                try {
                    TimeUnit.SECONDS.sleep(10L);
                } catch (Exception e) {
                    if (!XQRegistryClient.this.registryThreadStop) {
                        XQRegistryClient.log.error("注册线程异常.", e);
                    }
                }
            }
            XQRegistryClient.log.info("注册线程停止运行.");
            }
        });
        this.registryThread.setName("XQRegistryClient注册线程.");
        this.registryThread.setDaemon(true);
        this.registryThread.start();


        this.discoveryThread = new Thread(new Runnable() {
            public void run() {
                while (!XQRegistryClient.this.registryThreadStop) {
                    if (XQRegistryClient.this.discoveryData.size() == 0) {
                        try {
                            TimeUnit.SECONDS.sleep(3L);
                        } catch (Exception e) {
                            if (!XQRegistryClient.this.registryThreadStop) {
                                XQRegistryClient.log.error("客户端检索线程异常.", e);
                            }
                        }
                    } else {
                        try {
                            boolean monitorRet = XQRegistryClient.this.registryBaseClient.monitor(XQRegistryClient.this.discoveryData.keySet());
                            if (!monitorRet) {
                                TimeUnit.SECONDS.sleep(10L);
                            }
                            XQRegistryClient.this.refreshDiscoveryData(XQRegistryClient.this.discoveryData.keySet());
                        } catch (Exception e) {
                            if (!XQRegistryClient.this.registryThreadStop) {
                                XQRegistryClient.log.error("客户端检索线程异常.", e);
                            }
                        }
                    }
                }
                XQRegistryClient.log.info("检索线程停止");
            }
        });
        this.discoveryThread.setName("注册客户端检索线程.");
        this.discoveryThread.setDaemon(true);
        this.discoveryThread.start();

        log.info("XQRegistryClient初始化成功!");
    }

    public void stop() {
        this.registryThreadStop = true;
        if (this.registryThread != null) {
            this.registryThread.interrupt();
        }
        if (this.discoveryThread != null) {
            this.discoveryThread.interrupt();
        }
    }

    /**
     * 注册消费者
     * @param registryDataList
     * @return
     */
    public boolean registry(List<XQRegistryDataParamVO> registryDataList) {
        if ((registryDataList == null) || (registryDataList.size() == 0)) {
            throw new RuntimeException("消息队列注册 registryDataList为空");
        }
        for (XQRegistryDataParamVO registryParam : registryDataList) {
            if ((registryParam.getKey() == null) || registryParam.getKey().trim().length()==0) {
                throw new RuntimeException("消息队列注册 registryDataList Key为空");
            }
            if ((registryParam.getValue() == null) || registryParam.getValue().trim().length()==0) {
                throw new RuntimeException("消息队列注册 registryDataList 值为空");
            }
        }
        this.registryData.addAll(registryDataList);
        this.registryBaseClient.registry(registryDataList);
        return true;
    }

    public boolean remove(List<XQRegistryDataParamVO> registryDataList) {
        if ((registryDataList == null) || (registryDataList.size() == 0)) {
            throw new RuntimeException("消息队列注册 registryDataList为空");
        }
        for (XQRegistryDataParamVO registryParam : registryDataList) {
            if ((registryParam.getKey() == null) || registryParam.getKey().trim().length()==0) {
                throw new RuntimeException("消息队列注册 registryDataList Key为空");
            }
            if ((registryParam.getValue() == null) || registryParam.getValue().trim().length()==0) {
                throw new RuntimeException("消息队列注册 registryDataList 值为空");
            }
        }
        this.registryData.removeAll(registryDataList);


        this.registryBaseClient.remove(registryDataList);

        return true;
    }

    /**
     * 检索消费者信息
     * @param keys
     * @return
     */
    public Map<String, TreeSet<String>> discovery(Set<String> keys) {
        if ((keys == null) || (keys.size() == 0)) {
            return null;
        }
        Map<String, TreeSet<String>> registryDataTmp = new HashMap();
        for (String key : keys) {
            TreeSet<String> valueSet = (TreeSet) this.discoveryData.get(key);
            if (valueSet != null) {
                registryDataTmp.put(key, valueSet);
            }
        }
        if (keys.size() != registryDataTmp.size()) {
            refreshDiscoveryData(keys);
            for (String key : keys) {
                TreeSet<String> valueSet = (TreeSet) this.discoveryData.get(key);
                if (valueSet != null) {
                    registryDataTmp.put(key, valueSet);
                }
            }
        }
        return registryDataTmp;
    }

    /**
     * 检索注册消费者
     * @param keys
     */
    private void refreshDiscoveryData(Set<String> keys) {
        if ((keys == null) || (keys.size() == 0)) {
            return;
        }
        Map<String, TreeSet<String>> updatedData = new HashMap();

        Map<String, TreeSet<String>> keyValueListData = this.registryBaseClient.discovery(keys);
        if (keyValueListData != null) {
            for (String keyItem : keyValueListData.keySet()) {
                TreeSet<String> valueSet = new TreeSet();
                valueSet.addAll((Collection) keyValueListData.get(keyItem));


                boolean updated = true;
                TreeSet<String> oldValSet = (TreeSet) this.discoveryData.get(keyItem);
                if ((oldValSet != null) && (BasicJson.toJson(oldValSet).equals(BasicJson.toJson(valueSet)))) {
                    updated = false;
                }
                if (updated) {
                    this.discoveryData.put(keyItem, valueSet);
                    updatedData.put(keyItem, valueSet);
                }
            }
        }
        if (updatedData.size() > 0) {
            log.info("刷新检索数据完成, discoveryData(updated) = {}", updatedData);
        }
        log.debug("刷新检索数据完成, discoveryData = {}", this.discoveryData);
    }

    public TreeSet<String> discovery(String key) {
        if (key == null) {
            return null;
        }
        Map<String, TreeSet<String>> keyValueSetTmp = discovery(new HashSet(Arrays.asList(new String[]{key})));
        if (keyValueSetTmp != null) {
            return (TreeSet) keyValueSetTmp.get(key);
        }
        return null;
    }
}
